package com.example.scala.transformation

import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._

object Main {

  def map1(value:String):String = {
    value
  }

  def map2(value:Int):String = {
    value.toString + "map2"
  }

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val data1: DataStream[String] = env.fromElements(1, 2, 3, 4, 5, 6).map(("data1: " + _ ))
    val data2: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5, 6)

    val connected: ConnectedStreams[String, Int] = data1.connect(data2)
    val res: DataStream[String] = connected.map(new CoMapFunction[String, Int, String] {
      override def map1(value: String): String = {
        value
      }

      override def map2(value: Int): String = {
        value.toString + "map2"
      }
    })
    val res2: DataStream[String] = connected.map(v1 => v1,
      v2 => v2.toString + "map")
//    connected.map(map1(_),map2(_))


//    res.print
//    res2.print

    // T => TraversableOnce[String]
    val splited: SplitStream[Int] = data2.split(value => {
      val lst = Nil
      if (value % 2 == 0)
        "even" :: lst
      else
        "odd" :: lst
    })

    val evens: DataStream[Int] = splited.select("even")
    val odds: DataStream[Int] = splited.select("odd")
//    evens.print()
    odds.print()

    env.execute



  }

}
